package de.ruedigermoeller.fastcast.service.fcdemon;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import de.ruedigermoeller.fastcast.config.FCClusterConfig;
import de.ruedigermoeller.fastcast.config.FCTopicConf;
import de.ruedigermoeller.fastcast.remoting.*;
import de.ruedigermoeller.fastcast.service.FCMembership;
import de.ruedigermoeller.fastcast.util.FCLog;
import de.ruedigermoeller.fastcast.util.Sleeper;
import org.codehaus.janino.JavaSourceClassLoader;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA.
 * User: ruedi
 * Date: 9/5/13
 * Time: 8:49 PM
 * To change this template use File | Settings | File Templates.
 */
public class FCControl {

    @Parameter( names = { "-deploy", "-d" }, description = "deploy from default image ./control/cluster/deployed")
    Boolean deploy = false;

    @Parameter( names = { "-deployDir", "-dd" }, description = "deploy from given directory")
    String baseDir = null;

    @Parameter( names = { "-terminateAll", "-ta" }, description = "hard terminate all remote processes")
    Boolean terminateAll = false;

    @Parameter( names = { "-terminate", "-t" }, description = "hard terminate remote process with given (virtual) pid")
    String terminate = null;

    @Parameter( names = { "-j" }, variableArity = true, description = "run java file directly from source with janino. Expects sourcepath to start in 'control/cluster' and classes/jar's in 'cluster/deployed/lib' ")
    List<String> janinoCmd;

    @Parameter( names = { "-clean", "-cl" }, description = "completely clean shared dir of demons")
    Boolean clean = false;

    @Parameter( names = { "-r", "-restart" }, variableArity = true, description = "same as -ta -cl -d -j [ClassName]")
    List<String> janinoCmdRestart;

    @Parameter( names = { "-cluster", "-c" }, description = "defines name of cluster. if ommited read from conf or default name" )
    String clusterName;

    @Parameter( names = "-h", description = "defines homedir of fccontrol")
    String home = "./";

    @Parameter( names = { "-execRand", "-er" }, description = "run a process remotely on a random node" )
    String cmdLine;

    @Parameter( names = { "-exec", "-e" }, arity = 2, description = "run a process remotely on given node or host" )
    List<String> processSpec;

    @Parameter( names = { "-list", "-l"}, description =  "list processes" )
    Boolean list = false;

    @Parameter( names = { "-help", "-?", "?", "/?", "--help"}, description =  "print this" )
    Boolean usage = false;

    @Parameter( names = { "-v", "-verbose"}, description =  "verbose mode" )
    Boolean verbose = false;

    @Parameter( names = "-sendBench", description = "test max remote calls per second with current limit rate configuration" )
    Boolean sendBench = false;

    @Parameter( names = "-reqRespBench", description = "test max remote calls with response per second. (depends on maxOpenRespCalls property)" )
    Boolean sendReqRespBench = false;

    @Parameter( names = "-reqRespLat", description = "test roundtrip latency with request/response message with given rate per second" )
    Integer reqReqLat = -1;

    @Parameter( names = "-updateSelf", description = "update fastcast jars and terminate" )
    Boolean updateSelf = false;

    @Parameter( names = {"-printMemberStats", "-ps"},variableArity = true, arity = 2, description = "args: cluster.yaml [topicname|'all'] [NodeName (evaluated with 'startsWith')] [interval] tries to join FCMemberShip service of given cluster.yaml. if topic is omitted just lists node names. topic = 'all' lists stats of all topics. \nExample: fcon -ps bigtest.yaml rqserver Processor 1, fcon -ps bigtest.yaml" )
    List<String> yamlnamePlusStats;

    FCDemon remote;

    public FCDemon getRemote() {
        return remote;
    }

    public void setRemote(FCDemon remote) {
        this.remote = remote;
    }

    String findAnyClusterConfig() {
        String[] list1 = new File(home + File.separator + "cluster" + File.separator + "deployed").list();
        for (int i = 0; list1 != null && i < list1.length; i++) {
            String s = list1[i];
            if ( s.endsWith(".yaml") ) {
                return s;
            }
        }
        return null;
    }

    FCClusterConfig loadConfig( String finam ) throws IOException {
        return FCClusterConfig.read(home+File.separator+"cluster"+File.separator+"deployed"+File.separator+finam);
    }

    void dumpMemberShip(String yamlFinam, String topicName, final String nodeStartsWith, long millis ) {
        if ( yamlFinam == null ) {
            yamlFinam = findAnyClusterConfig();
            if ( yamlFinam == null ) {
                System.out.println("no clusterconfig found in ./cluster/deployed");
                System.exit(0);
            }
            System.out.println("defaulting to cluster config "+yamlFinam);
        }
        try {
            FCTopicConf toJoin = null;
            FCClusterConfig fcClusterConfig = loadConfig(yamlFinam);
            FCTopicConf[] topics = (FCTopicConf[]) fcClusterConfig.getTopics();
            for (int i = 0; i < topics.length; i++) {
                FCTopicConf topic = topics[i];
                if ( topic.getServiceClass().equals(FCMembership.class.getName()) ) {
                    toJoin = topic;
                    break;
                }
            }
            if ( toJoin != null ) {
                FastCast.getRemoting().joinCluster(fcClusterConfig,"fcon",null);
                if ( ! toJoin.isAutoStart() ) {
                    FastCast.getRemoting().start(toJoin.getName());
                }
                FCMembership membership = (FCMembership) FastCast.getRemoting().getService(toJoin.getName());
                if ("all".equals(topicName) || topicName != null ) {
                    do {
                        FCMembership remoteMembership = (FCMembership) FastCast.getRemoting().getRemoteService(toJoin.getName());
                        for (int i = 0; i < topics.length; i++) {
                            final FCTopicConf topic = topics[i];
                            if ("all".equals(topicName) || topicName.equals(topic.getName()) ) {
                                remoteMembership.getStats(topic.getName(), new FCFutureResultHandler() {
                                    @Override
                                    public void resultReceived(Object obj, String sender) {
                                        if ( nodeStartsWith == null || sender.startsWith(nodeStartsWith) )
                                        {
                                            System.out.println("Stats from sender "+sender+" on topic "+topic.getName() );
                                            System.out.println(obj);
                                            System.out.println("----");
                                        }
                                    }
                                });
                            }
                        }
                        if ( millis > 0 )
                            Thread.sleep(millis);
                    } while ( millis > 0 );
                }
                Thread.sleep(3000);
                if ( topicName == null ) {
                    System.out.println("");
                    System.out.println("Members:");
                    System.out.println(membership.dumpToString());
                    System.out.println("topics:");
                    for (int i = 0; i < topics.length; i++) {
                        final FCTopicConf topic = topics[i];
                        System.out.println("  "+topic.getName());
                    }
                }
            } else {
                System.out.println("no FCMemberShip topic found");
            }
        } catch (Exception e) {
            System.out.println(e.getMessage());
            System.out.println("failed to load config");
            e.printStackTrace();
        }
    }


    private List<FCDemon.NodeInfo> countReceiver() {
        if ( verbose )
            System.out.println("counting receivers ..");
        final Semaphore sema = new Semaphore(0);
        final ArrayList<FCDemon.NodeInfo> res = new ArrayList<>();
        remote.getProcessList(new FCFutureResultHandler() {
            @Override
            public void resultReceived(Object obj, String sender) {
                synchronized (res) {
                    res.add((FCDemon.NodeInfo) obj);
                }
                if (verbose)
                    System.out.println("demon found " + obj);
                sema.release();
            }
        });
        try {
            sema.tryAcquire(5000,TimeUnit.MILLISECONDS);
            Thread.sleep(500); // wait for additional responses
        } catch (InterruptedException e) {
            e.printStackTrace();
            FCLog.log(e);
        }
        return res;
    }

    private void terminate() {
        if ( verbose )
            System.out.println("sending terminate "+terminate);
        final CountDownLatch latch = new CountDownLatch(1);
        remote.killProc(terminate, new FCFutureResultHandler() {
            @Override
            public void resultReceived(Object obj, String sender) {
                System.out.println(obj);
                latch.countDown();
            }
        });
        try {
            latch.await(1000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {

        }
    }

    public void terminateAll() {
        if ( verbose )
            System.out.println("sending kill ..");
        remote.killAllProcesses();
    }

    public void clean() {
        if ( verbose )
            System.out.println("sending clean ..");
        remote.delBaseDir();
    }

    private void sync(int numReceivers ) {
        final CountDownLatch sem = new CountDownLatch(numReceivers+1);
        remote.sync(new FCFutureResultHandler() {
            @Override
            public void timeoutReached() {
                if ( sem.getCount() != 1 ) {
                    System.out.println(" Error: "+(sem.getCount()-1)+" demons did not aknowledge !");
                }
                if ( errors.get() > 0 ) {
                    if ( baseDir != null )
                        System.out.println( errors.get()+" errors during copy.");
                    else
                        System.out.println( errors.get()+" errors occured.");
                } else
                    System.out.println("[all demons acknowledged]");
                while( sem.getCount() > 0 )
                    sem.countDown();
            }

            @Override
            public void resultReceived(Object obj, String sender) {
                sem.countDown();
                if (sem.getCount()==1)
                    timeoutReached();
                if ( verbose )
                    System.out.println("sync received by "+ FCReceiveContext.get().getSender());
            }
        });
        try {
            sem.await(60000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            FCLog.log(e);
        }
    }

    AtomicInteger errors = new AtomicInteger(0);

    private void deploy(final int receivers) throws IOException {
        File file = new File(baseDir);
        final File baseFile = file.getCanonicalFile();
        try {
            Files.walkFileTree(FileSystems.getDefault().getPath(baseDir), new FileVisitor<Path>() {
                @Override
                public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                    return FileVisitResult.CONTINUE;
                }

                @Override
                public FileVisitResult visitFile(Path file0, BasicFileAttributes attrs) throws IOException {
                    final File file = file0.toFile().getCanonicalFile();
                    byte cont[] = new byte[(int) file.length()];
                    FileInputStream in = new FileInputStream(file);
                    try {
                        in.read(cont);
                        String relDir0 = file.getParentFile().getAbsolutePath().substring(baseFile.getAbsolutePath().length());
                        if (updateSelf) {
                            relDir0 = "$LIB"+relDir0;
                        }
                        final String relDir = relDir0;
                        final String name = file.getName();

                        if ( verbose ) {
                            System.out.println("sending "+relDir+" "+name+" length "+cont.length/1000+"k");
                        }
                        remote.deploy(relDir, name, cont, new FCFutureResultHandler() {
                            @Override
                            public void timeoutReached() {
                                if ( recCount.get() != 0) {
                                    errors.incrementAndGet();
                                    System.out.println("Error: Not all receivers acknowledged copy of "+relDir+"/"+name);
                                }
                            }

                            AtomicInteger recCount = new AtomicInteger(receivers);
                            @Override
                            public void resultReceived(Object obj, String sender) {
                                if ( ! obj.equals(Boolean.TRUE) ) {
                                    System.out.println("error deploying file "+file+" in node "+FCReceiveContext.get().getSender());
                                    errors.incrementAndGet();
                                } else {
                                    recCount.decrementAndGet();
                                }
                                if (recCount.get()==0)
                                    done();
                            }
                        });
                        return FileVisitResult.CONTINUE;
                    } finally {
                        in.close();
                    }
                }

                @Override
                public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
                    return FileVisitResult.CONTINUE;
                }

                @Override
                public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                    return FileVisitResult.CONTINUE;
                }
            });
        } catch (IOException e) {
            FCLog.log(e);
        }
    }

    private void execute(List<FCDemon.NodeInfo> recs, String receiver /*choose randomly if null*/) {
        if ( recs .size() == 0 ) {
            System.out.println("Error. No demons found, can't execute");
            return;
        }
        final CountDownLatch latch = new CountDownLatch(1);
        if ( receiver == null )
            receiver = recs.get((int) (recs.size()*Math.random())).getNodeId();
        else {
            boolean found = false;
            for (int i = 0; i < recs.size(); i++) {
                FCDemon.NodeInfo nodeInfo = recs.get(i);
                String s = nodeInfo.getNodeId();
                if ( s.equals(receiver) ) {
                    found = true;
                    break;
                }
                if ( receiver.equals(nodeInfo.getHost()) ) {
                    found = true;
                    receiver = nodeInfo.getNodeId();
                    break;
                }
            }
            if ( ! found ) {
                System.out.println("Error: Receiver address not found in nodes");
                return;
            }
        }
        FCSendContext.get().setReceiver(receiver);
        int first = cmdLine.indexOf(" ");
        if ( first < 0)
            first = cmdLine.length();
        String name = cmdLine.substring(0, first);
        remote.startProcess(1, name, cmdLine, new FCFutureResultHandler() {
            @Override
            public void resultReceived(Object obj, String sender) {
                if ( obj instanceof FCDemon.ProcessEntry == false) {
                    System.out.println("error starting process:"+obj);
                } else {
                    System.out.println("started process "+obj);
                }
                latch.countDown();
            }
        });
        try {
            latch.await(3000,TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            FCLog.log(e);
        }
        if ( latch.getCount() > 0 ) {
            System.out.println("no response, process probably not started");
        }
    }

    private void list(int numNodes) {
        final CountDownLatch latch = new CountDownLatch(numNodes);
        remote.getProcessList(new FCFutureResultHandler() {
            @Override
            public void resultReceived(Object obj, String sender) {
                latch.countDown();
                FCDemon.NodeInfo info = (FCDemon.NodeInfo) obj;
                List<FCDemon.ProcessEntry> res = info.getProcesses();
                System.out.println("Sender " + info + " ");
                for (int i = 0; i < res.size(); i++) {
                    FCDemon.ProcessEntry processEntry = res.get(i);
                    if ( verbose )
                        System.out.println("    " + processEntry);
                    else {
                        System.out.println("    " + "ProcessEntry{" +
                                "pid='" + processEntry.getPid() + '\'' +'}');
                    }
                }
            }
        });
        try {
            latch.await(3000,TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            FCLog.log(e);
        }
        if ( latch.getCount() > 0 ) {
            System.out.println("Error: Not all receivers responded");
        }
    }

    private void doSendBench() {
        remote.maxMethodsCallSpeedReceiver(true);
        int count = 1;
        long lastMeasure = System.currentTimeMillis();
        while( true ) {
            remote.maxMethodsCallSpeedReceiver(false);
            count++;
            if ( count % 1000 == 0 ) {
                long now = System.currentTimeMillis();
                long diff = now - lastMeasure;
                if ( diff > 1000 ) {
                    System.out.println("sent "+count+" remote calls in "+diff+" ms");
                    count = 0; lastMeasure=now;
                }
            }
        }
    }

    private void doReqRespLatency(int sendRatePerSecond) {
        Sleeper sleeper = new Sleeper();
        int sleep = 1000*1000/sendRatePerSecond;
        final ConcurrentHashMap<String,Long> latMap = new ConcurrentHashMap<>();
        int count = 1;
        long lastMeasure = System.currentTimeMillis();
        while( true ) {
            sleeper.sleepMicros(sleep);
            remote.reqRespTest(System.nanoTime(), new FCFutureResultHandler() {
                @Override
                public void resultReceived(Object obj, String sender) {
                    Long tim = (Long) obj;
                    latMap.put(FCReceiveContext.get().getSender(), System.nanoTime() - tim);
                }
            });
            count++;
            if ( count % 100 == 0 ) {
                long now = System.currentTimeMillis();
                long diff = now - lastMeasure;
                if ( diff > 1000 ) {
                    System.out.println("sent "+count+" remote calls in "+diff+" ms.");
                    for (Iterator<String> iterator = latMap.keySet().iterator(); iterator.hasNext(); ) {
                        String receiver = iterator.next();
                        System.out.println("  "+receiver+" "+latMap.get(receiver)/1000+" micro seconds");
                    }
                    count = 0; lastMeasure = now;
                    latMap.clear();
                }
            }
        }
    }

    private void doReqRespThroughput() {
        int count = 1;
        final AtomicInteger reccount = new AtomicInteger(1);
        long lastMeasure = System.currentTimeMillis();
        while( true ) {
            final long nanosIn = System.nanoTime();
            remote.reqRespTest(nanosIn, new FCFutureResultHandler() {
                @Override
                public void resultReceived(Object obj, String sender) {
                    if (nanosIn != ((Long)obj).longValue() )
                        throw new RuntimeException("req resp error ! calback confusion");
                    reccount.incrementAndGet();
                    done(); // clear callback on first response
                }
            });
//            try {
//                Thread.sleep(1);
//            } catch (InterruptedException e) {
//                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
//            }
            count++;
            if ( count % 5000 == 0 )
            {
                long now = System.currentTimeMillis();
                long diff = now - lastMeasure;
                if ( diff > 1000 ) {
                    System.out.println("sent "+count+" remote calls with resull in "+diff+" ms. Received "+reccount.get()+" results");
                    count = 0; lastMeasure = now; reccount.set(0);
                }
            }
        }
    }

    void execJanino(List<String> janinoCmd, List<FCDemon.NodeInfo> receivers) {
        JavaSourceClassLoader loader = new JavaSourceClassLoader(getClass().getClassLoader(),new File[] {
           new File(home+File.separator+"cluster"),
        }, null);
        String arg = "";
        for (int i = 1; i < janinoCmd.size(); i++) {
            String s = janinoCmd.get(i);
            arg+=s+" ";
        }
        FCStartScript script = null;
        try {
            Class<?> aClass = loader.loadClass(janinoCmd.get(0));
            script = (FCStartScript) aClass.newInstance();
            script.setFchome(home);
            script.setReceivers(receivers);
            script.initCluster();
            script.startCluster(arg.trim(), receivers );
        } catch (Exception e) {
            e.printStackTrace();
        }
        if ( script.errors.size() > 0 ) {
            for (int i = 0; i < script.errors.size(); i++) {
                String s = (String) script.errors.get(i);
                System.out.println("Error:"+s);
            }
        }
    }

    void updateSelfJars(int count) throws IOException {
        baseDir = home+File.separator+"lib";
        deploy(count);
        getRemote().terminate();
    }

    public static void main(String arg[]) throws IOException, InterruptedException {
        FCControl com = new FCControl();
        JCommander jCommander = null;
        try {
            jCommander = new JCommander(com, arg);
        } catch (Exception e) {
            System.out.println(e.getMessage());
            return;
        }

        if ( com.usage ) {
            jCommander.usage();
            return;
        }

        if ( com.verbose ) {
            FCLog.get().setLogLevel(FCLog.INFO);
        }

        if ( com.yamlnamePlusStats != null ) {
            com.dumpMemberShip(
                    com.yamlnamePlusStats.size() > 0 ? com.yamlnamePlusStats.get(0) : null,
                    com.yamlnamePlusStats.size() > 1 ? com.yamlnamePlusStats.get(1) : null,
                    com.yamlnamePlusStats.size() > 2 ? com.yamlnamePlusStats.get(2) : null,
                    com.yamlnamePlusStats.size() > 3 ? Integer.parseInt(""+com.yamlnamePlusStats.get(3))*1000l : 0
            );
            System.exit(0);
        }

        FCRemoting rem = FastCast.getRemoting();
        rem.joinCluster(com.home+"/fccontrol.yaml", "FCControl", com.clusterName);
        rem.startSending("fcdemon");

        com.setRemote((FCDemon) rem.getRemoteService("fcdemon"));
        List<FCDemon.NodeInfo> receivers = com.countReceiver();
        System.out.println("found "+receivers.size()+" receivers ..");
        if ( com.updateSelf ) {
            com.terminateAll();
            com.updateSelfJars(receivers.size());
            Thread.sleep(2000);
            System.exit(-1);
        }
        if ( com.janinoCmdRestart != null ) {
            com.janinoCmd = com.janinoCmdRestart;
            com.terminateAll = true;
            com.clean = true;
            com.deploy = true;
        }
        if ( com.terminateAll) {
            com.terminateAll();
        }
        if ( com.terminate != null )
            com.terminate();
        if ( com.clean ) {
            com.clean();
        }
        if ( com.deploy && com.baseDir == null ) {
            com.baseDir = com.home+File.separator+"cluster"+File.separator+"deployed";
        }
        if ( com.baseDir != null ) {
            com.deploy(receivers.size());
        }
        if ( com.cmdLine != null ) {
            com.execute(receivers, null);
        }
        if ( com.processSpec != null ) {
            com.cmdLine = com.processSpec.get(0);
            com.execute(receivers, com.processSpec.get(1));
        }
        if ( com.list ) {
            com.list(receivers.size());
        }
        if ( com.sendBench ) {
            com.doSendBench();
        }
        if ( com.sendReqRespBench ) {
            com.doReqRespThroughput();
        }
        if ( com.reqReqLat > 0 ) {
            com.doReqRespLatency(com.reqReqLat);
        }
        if ( com.janinoCmd != null ) {
            com.execJanino(com.janinoCmd, receivers);
        }
        com.sync(receivers.size());
        System.exit(-1);
    }



}
